/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service.busidatawriter;

import com.asiainfo.redis.service.BusiConfigService;
import redis.clients.util.SafeEncoder;

import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @version v 1.0 on 2016/7/27 18:45
 * @auther william.xu
 */
public class BusiDataRedisWriter4TwoConditions extends BusiDataRedisWriter {
    public BusiDataRedisWriter4TwoConditions(Integer operFlag) {
        super(operFlag);
    }

    @Override
    public void doWrite(ResultSet rst) throws Exception {
        String busiTypeCode = busiTypeRepository.findOne(busiDataRule.getBusiTypeId()).getBusiCode();
        String keyPrefix = busiDataRule.getKeyPrefix();

        //数据集中会根据ConditionField1,ConditionField2字段的值排序
        String conditionField1 = busiDataRule.getConditionField1();
        String conditionField2 = busiDataRule.getConditionField2();

        List<Map> dataList = new ArrayList();
        String[] saveFields = busiDataRule.getSaveFields().split(BusiConfigService.SPLIT_CODE);
        String oldConditionFieldValue1 = null;
        String oldConditionFieldValue2 = null;
        byte[] key = null;
        while (rst.next()) {
            Object value1 = rst.getObject(conditionField1);
            String conditionFieldValue1 = (value1==null?"":value1.toString());
            Object value2 = rst.getObject(conditionField2);
            String conditionFieldValue2 = (value2==null?"":value2.toString());
            //读取第一行时
            if (oldConditionFieldValue1 == null){
                oldConditionFieldValue1 = conditionFieldValue1;
                key = SafeEncoder.encode(busiTypeCode+":"+keyPrefix+":"+oldConditionFieldValue1);
                connection.del(key);
                oldConditionFieldValue2 = conditionFieldValue2;
            }
            //ConditionField1字段的值开始不一样时
            if (conditionFieldValue1 != null && !conditionFieldValue1.equals(oldConditionFieldValue1)){
                writetoRedis(key, dataList, oldConditionFieldValue2);
                //重置oldValue1和key的值
                oldConditionFieldValue1 = conditionFieldValue1;
                key = SafeEncoder.encode(busiTypeCode+":"+keyPrefix+":"+oldConditionFieldValue1);
                connection.del(key);
                //清空之前的list，重置oldValue2的值
                dataList = new ArrayList();
                oldConditionFieldValue2 = conditionFieldValue2;
            }
            if (conditionFieldValue2 != null && !conditionFieldValue2.equals(oldConditionFieldValue2)){
                ////ConditionField2字段的值开始不一样了，需要把之前取到的数据写入redis的hash结构
                writetoRedis(key, dataList, oldConditionFieldValue2);
                //清空之前的list，重置oldValue2的值
                dataList = new ArrayList();
                oldConditionFieldValue2 = conditionFieldValue2;
            }
            //把取到的数据加到list中
            Map row = new HashMap();
            for (String saveField : saveFields) {
                row.put(saveField, rst.getObject(saveField));
            }
            dataList.add(row);
        }
        //遍历结束，把最后一段数据写入redis
        if (key != null) {
            writetoRedis(key, dataList, oldConditionFieldValue2);
        }
    }

    private void writetoRedis(byte[] key, List<Map> dataList,  String oldConditionFieldValue2) throws Exception{
        if (operFlag == 0) {
            connection.hSet(key, SafeEncoder.encode(oldConditionFieldValue2), SafeEncoder.encode(object2String(dataList)));
        }
        publishMessage(key);
    }

}
